package com.atguigu.gmall.realtime.common.util;

import com.atguigu.gmall.realtime.common.constant.Constant;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;

import java.io.IOException;
import java.nio.charset.StandardCharsets;

/**
 * Flink Source 获取的工具类, 提供各种FlinkSource对象的获取
 */
public class FlinkSourceUtil {
    public static MySqlSource<String> getMySqlSource(String database, String table){
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_PORT)
                .databaseList(database) // 设置捕获的数据库， 如果需要同步整个数据库，请将 tableList 设置为 ".*".
                .tableList(database+"."+table) // 设置捕获的表
                .username(Constant.MYSQL_USER_NAME)
                .password(Constant.MYSQL_PASSWORD)
                .deserializer(new JsonDebeziumDeserializationSchema()) // 将 SourceRecord 转换为 JSON 字符串
                .startupOptions(StartupOptions.initial())
                /*配置选项scan.startup.mode指定 MySQL CDC 使用者的启动模式。有效枚举包括：
                    initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。
                    earliest-offset：跳过快照阶段，从可读取的最早 binlog 位点开始读取
                    latest-offset：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
                    specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
                    timestamp：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。
                * */
                .build();
                return mySqlSource;
    }

    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        // 3.创建KafkaSource对象, 从Kafka中读取数据
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers(Constant.KAFKA_BROKERS)
                .setTopics(topic)
                .setGroupId(groupId)
                .setValueOnlyDeserializer(
                        // 不能处理null值,遇到null值会报错
                        // new SimpleStringSchema()

                        // 自己完成反序列化的操作
                        new DeserializationSchema<String>() {
                            @Override
                            public String deserialize(byte[] message) throws IOException {
                                // 判断是否消费到null
                                if(message != null){
                                    return new String(message, StandardCharsets.UTF_8);
                                }

                                return null;
                            }

                            @Override
                            public boolean isEndOfStream(String s) {
                                return false;  // 数据是否读到尾了? false,流式处理没有尾
                            }

                            @Override
                            public TypeInformation<String> getProducedType() {
                                return BasicTypeInfo.STRING_TYPE_INFO;
                            }
                        }

                )
                // .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                // 为了方便开发过程中的各种测试, 直接将offset重置到头, 方便读取到主题中已有的数据, 不需要每次都模拟生成新的数据
                .setStartingOffsets(OffsetsInitializer.latest())
                // .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")  隔离级别,读已提交
                .build();
        return kafkaSource;
    }
}
